package com.zshield.util;

import com.zshield.config.KafkaConfig;
import com.zshield.run.KafkaPreCompute;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.RequestLine;
import org.apache.http.client.config.RequestConfig;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;
import org.apache.log4j.chainsaw.Main;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequest;
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.admin.indices.open.OpenIndexRequest;
import org.elasticsearch.action.admin.indices.open.OpenIndexResponse;
import org.elasticsearch.client.*;
import org.elasticsearch.client.transport.TransportClient;
import org.elasticsearch.cluster.metadata.IndexMetaData;
import org.elasticsearch.cluster.metadata.RepositoriesMetaData;
import org.elasticsearch.common.transport.TransportAddress;

import java.io.IOException;
import java.io.InputStream;
import java.util.*;

import static org.elasticsearch.client.RestClientBuilder.DEFAULT_SOCKET_TIMEOUT_MILLIS;

public class ESclient {
    private static final Logger logger = Logger.getLogger(ESclient.class);

    private static RestHighLevelClient highClient = null;

    private ESclient(){}

    public static RestHighLevelClient getHighClient(){
        if (highClient == null){
            synchronized (ESclient.class){
                if (highClient == null){
                    HttpHost host = new HttpHost(KafkaPreCompute.es_host,9200,"http");
                    RestClientBuilder builder = RestClient.builder(host).setRequestConfigCallback(new RestClientBuilder.RequestConfigCallback() {
                        @Override
                        public RequestConfig.Builder customizeRequestConfig(RequestConfig.Builder requestConfigBuilder) {
                            requestConfigBuilder.setSocketTimeout(DEFAULT_SOCKET_TIMEOUT_MILLIS * 4);
                            return requestConfigBuilder;
                        }
                    }).setMaxRetryTimeoutMillis(300000);
                    builder.build();
                    highClient = new RestHighLevelClient(builder);
                }
            }
        }
        return highClient;
    }

    public static String[] listEsIndex() {
        try {
            highClient = getHighClient();
            Response response = highClient.getLowLevelClient().performRequest("GET", "/_cat/indices");
            HttpEntity entity = response.getEntity();
            byte[] bytes = entity.getContent().readAllBytes();
            String content = new String(bytes, "utf-8");
            String[] indices = content.split("\n");
            Arrays.sort(indices);
            for (int i = 0;i < indices.length; i++ ) {
                indices[i] = indices[i].split("\\s+")[2];
                logger.info("list es index:[" + indices[i] + "]");
            }
            return indices;
        } catch (Exception e) {
            logger.error("[list index exception] [The reason for error {" + e + "}]");
            KafkaConfig.printErrorLog(logger,e);
            System.exit(1);
        }
        return null;
    }

    public static void deleteEsIndex(String index) {
        try {
            highClient = getHighClient();
            Response response = highClient.getLowLevelClient().performRequest("DELETE", index);
        } catch (IOException e) {
            logger.error("[delete index exception] [The reason for error {" + e + "}]");
            KafkaConfig.printErrorLog(logger,e);
            System.exit(1);
        }
    }

}

